package cn.piesat.scanning.business.dq1043.single;

import cn.piesat.scanning.business.dq1043.vo.ViewRecordVo;
import cn.piesat.scanning.service.*;
import cn.piesat.scanning.utils.RedisUtil;
import com.alibaba.fastjson.JSON;
import cn.piesat.scanning.business.common.algorithm.processAlgorithm.SingleTaskAbstract;
import cn.piesat.scanning.business.dq1043.vo.RuleVO;
import cn.piesat.scanning.dto.*;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
import redis.clients.jedis.Jedis;

import java.io.File;
import java.sql.Date;
import java.text.SimpleDateFormat;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * @author xhl
 */
@Component
public class SingleTask extends SingleTaskAbstract {

    @Value("${scanning.data.day}")
    private Integer nearDay;

    private final DateTimeFormatter TIME_FORMAT =  DateTimeFormatter.ofPattern("yyyyMMddHHmmss");

    private static final String NULL = "NULL";
    /**
     * 同一谱段不同通道的时间差
     */
    private static final int DIFF_TIME_MINUTE = 3;

    private static final Logger logger = LoggerFactory.getLogger(SingleTask.class);

    @Autowired
    private HtScanningMetadataService htScanningMetadataService;
    @Autowired
    private DbDmsSchedulerTaskFileRecordService dbDmsSchedulerTaskFileRecordService;
    @Autowired
    private HtPathConfigurationService htPathConfigurationService;
    @Autowired
    private DbDmsScanningMetadataAuxInfoService dbDmsScanningMetadataAuxInfoService;
    @Autowired
    private DbDmsScanningValidationService dbDmsScanningValidationService;
    @Autowired
    private DbDmsSchedulerTaskRecordLogService dbDmsSchedulerTaskRecordLogService;
    @Autowired
    private DbDmsSchedulerTaskFileRecordFlowService dbDmsSchedulerTaskFileRecordFlowService;
    @Autowired
    private DbDmsFlowInfoExecService dbDmsFlowInfoExecService;
    @Autowired
    private DbDmsFlowInfoExecStatusService dbDmsFlowInfoExecStatusService;

    @Autowired
    private DbDmsFlowInfoExecIdFlowIdService dbDmsFlowInfoExecIdFlowIdService;

    @Autowired
    private RedisTemplate<String, String> redisTemplate;

    @Autowired
    private HtAlgorithmConfigurationService htAlgorithmConfigurationService;
    @Autowired
    private KafkaTemplate<String,String> kafkaTemplate;

    @Value("${spring.kafka.producer.dmstaskexec.topic}")
    private String execTopic;

    @Autowired
    private HtScheduleJobService htScheduleJobService;

    @Override
    public void parseRuleAndFindData(Map<String, Object> inputParams, Map<String, Object> ruleParams, Map<String, Integer> indexParams, HtScheduleJobDTO taskDTO) {
        // TODO Auto-generated method stub
    }

    @Override
    public void parseRuleAndFindData(Map<String, Object> inputParams, Map<String, Object> ruleParams, HtScheduleJobDTO taskDTO) {
        AlgorithmTaskRuleParam taskRuleParam = JSON.parseObject(JSON.toJSONString(ruleParams), AlgorithmTaskRuleParam.class);
        AlgorithmTaskPathParam taskPathParam = JSON.parseObject(JSON.toJSONString(inputParams), AlgorithmTaskPathParam.class);
        //判断规则不为NULL但是输入路径为空的情况
        if (!rulePathExist(taskRuleParam,taskPathParam)) {
            return;
        }
        String preProcessRuleStr = taskRuleParam.getPreprocess();
        RuleVO preProcessRuleVO = JSON.parseObject(preProcessRuleStr, RuleVO.class);
        Boolean preProcess = null;
        try {
            preProcess = JSON.parseObject(preProcessRuleVO.getMatchRule(),Boolean.class);
        } catch (Exception e) {
            logger.error("标准化处理配置错误！");
            return ;
        }
        //TODO 暂时只满足1043 所有的主文件都一样
        RuleVO primaryRuleVO = JSON.parseObject(taskRuleParam.getPrimaryFile(), RuleVO.class);
        //查找可执行的数据，默认查找配置文件中配置的时间文件，[状态为重做的文件]，记录表中不存在的主文件
        List<HtScanningMetadataDTO> execMetadataList = new ArrayList<>();
        execMetadataList.addAll(selectExecMainFileList(primaryRuleVO,preProcess,taskDTO));
        execMetadataList.addAll(findRedoMetadataList(taskDTO.getId()));
        assemble(taskRuleParam,taskPathParam,execMetadataList,taskDTO);
    }

    private void assemble(AlgorithmTaskRuleParam taskRuleParam, AlgorithmTaskPathParam taskPathParam,
                          List<HtScanningMetadataDTO> execMetadataList,HtScheduleJobDTO taskDTO) {
        HtPathConfigurationDTO pathDTO = htPathConfigurationService.findById(taskDTO.getPathConfigId());
        RuleVO primaryRuleVO = JSON.parseObject(taskRuleParam.getPrimaryFile(), RuleVO.class);
        String[] allMode = primaryRuleVO.getMatchRule().split(",");
        String preProcessRuleStr = taskRuleParam.getPreprocess();
        RuleVO preProcessRuleVO = JSON.parseObject(preProcessRuleStr, RuleVO.class);
        Boolean preProcess = null;
        try {
            preProcess = JSON.parseObject(preProcessRuleVO.getMatchRule(),Boolean.class);
        } catch (Exception e) {
            logger.error("标准化处理配置错误！");
            return ;
        }
        for (HtScanningMetadataDTO metadataDTO : execMetadataList) {
            Map<String,Object> execParams = new HashMap<>();
            //匹配primaryFile规则
            String primaryFile = null;
            if(!("NULL").equals(primaryRuleVO.getProductIdentify())&& StringUtils.isNotBlank(primaryRuleVO.getProductIdentify())){
                primaryFile =pathDTO.getScanningPath()+ File.separator + metadataDTO.getFilePath();
            }else{
                for (String mode : allMode) {
                    primaryFile = primaryRule(primaryRuleVO, preProcess, metadataDTO, pathDTO, mode);
                    if (primaryFile != null) {
                        logger.debug("=============================");
                        logger.debug("--[role-主文件]--:找到波段数据集合{}",primaryFile);
                        break;
                    }
                }
                if (primaryFile == null) {
                    continue;
                }
            }

            //匹配GEO规则
            Map<String,String> geoOrRpbFile = geoRule(taskRuleParam.getGeoFile(),preProcess,metadataDTO,taskPathParam.getGeoFile());
            if (geoOrRpbFile == null) {
                logger.debug("--[role-geo]--:没有找到geo文件!");
                continue;
            }
            //匹配L2文件规则
            String l2File = l2Rule(taskRuleParam.getL2File(), metadataDTO, taskPathParam.getL2File());
            if (l2File == null) {
                logger.debug("--[role-L2]--:没有找到L2文件!");
                continue;
            }
            //匹配T799文件规则
            String nwpFile = nwpRule(taskRuleParam.getNwpFile(), metadataDTO.getFileDate(), taskPathParam.getNwpFile());
            if (nwpFile == null) {
                logger.debug("--[role-nwpFile]--:没有找到t799数据!");
                continue;
            }
            //匹配质检源文件数据规则
            String validationFile = validationRule(taskRuleParam.getValidationFile(), metadataDTO.getFileDate(), taskPathParam.getValidationFile());
            if (validationFile == null) {
                logger.debug("--[role-validationFile]--:没有找到检验源数据!");
                continue;
            }else if(StringUtils.isNotBlank(validationFile)){
                //质检算法需要加上主文件productIdentify和算法类型
                execParams.put("productIdentify",primaryRuleVO.getProductIdentify());
                execParams.put("algorithmType","validation");
            }
            logger.debug("-----主文件{}及相关算法文件扫描完毕-----",metadataDTO.getFileName());
            String validationName = "";
            if(StringUtils.isNotBlank(taskRuleParam.getValidationName())){
                RuleVO ruleVO = JSON.parseObject(taskRuleParam.getValidationName(),RuleVO.class);
                validationName = ruleVO.getMatchRule();
            }
            execParams.put("preprocess",preProcess);
            execParams.put("algorithmId",taskDTO.getAlgoId());
            execParams.put("pathConfigId",taskDTO.getPathConfigId());
            execParams.put("fileDate",metadataDTO.getFileDate());
            execParams.put("lon",metadataDTO.getLon());
            execParams.put("lat",metadataDTO.getLat());
            execParams.put("taskPlanId",metadataDTO.getTaskPlanId());
            execParams.put("subPlanId",metadataDTO.getSubPlanId());
            execParams.put("viewNumber",metadataDTO.getViewNumber());
            execParams.put("productDir",pathDTO.getOutputPath());
            Map<String,Object> algorithmParams = new HashMap<>();

            algorithmParams.put("primaryFile",primaryFile);
            if(StringUtils.isNotBlank(nwpFile)){
                algorithmParams.put("nwpFile",nwpFile);
            }
            algorithmParams.put("geoFile",geoOrRpbFile.get("geo"));
            algorithmParams.put("rpbFile",geoOrRpbFile.get("rpb"));
            algorithmParams.put("L2File",l2File);
            algorithmParams.put("tcfPath",taskPathParam.getTcfPath());
            algorithmParams.put("resultPath",pathDTO.getOutputPath());
            algorithmParams.put("auxPath",pathDTO.getAssistPath());
            algorithmParams.put("mainFileName",metadataDTO.getFileName());
            algorithmParams.put("validationFile",validationFile);
            algorithmParams.put("validationName",validationName);
            updateRecordAndSendMessageInsertFlow(execParams,taskDTO,metadataDTO.getFileName(),algorithmParams);
        }
    }

    //检验源文件规则解析
    private String validationRule(String validationFile, String fileDate, String validationFile1) {
        StringBuilder filePathSb = new StringBuilder();
        if (StringUtils.isBlank(validationFile)) {
            return filePathSb.toString();
        }
        RuleVO ruleVO = JSON.parseObject(validationFile, RuleVO.class);
        if (NULL.equals(ruleVO.getMatchRule()) || NULL.equals(ruleVO.getLevel())) {
            return filePathSb.toString();
        }
        fileDate = TIME_FORMAT.format(LocalDateTime.parse(fileDate, TIME_FORMAT).plusHours(-8L));
        Map<String,String> timeRangeParam = parseTimeRange(fileDate, ruleVO.getTimeRange());
        String mathexp = ruleVO.getMatchRule();
        String satellite = ruleVO.getSatellite();
        String startTime = timeRangeParam.get("startTime");
        String endTime = timeRangeParam.get("endTime");
        List<DbDmsScanningValidationDTO> ruleFileList = dbDmsScanningValidationService
                .findBySatelliteAndDataLevelAndProductIdentifyAndFileDateBetween(satellite,ruleVO.getLevel(),mathexp,startTime,endTime);
        if (ruleFileList.isEmpty()) {
            logger.info("未查询到任何质检源数据文件");
            return null;
        }
        for (DbDmsScanningValidationDTO fileDTO : ruleFileList) {
            String currentFilePath = validationFile1 + File.separator + fileDTO.getFilePath();
            File file = new File(currentFilePath);
            if (!file.exists()) {
                logger.info( "{}在磁盘上不存在",fileDTO.getFileName());
                return null;
            }
            filePathSb.append(",").append(currentFilePath);
        }

        return filePathSb.substring(1);
    }

    /**
     *
     * @param inputParams 输入参数
     * @param ruleParams 规则参数
     * @param taskDTO 任务信息对象
     * @param data 处理的数据对象，自己定义
     */
    @Override
    public void parseRuleAndFindData(Map<String, Object> inputParams, Map<String, Object> ruleParams,
                                     HtScheduleJobDTO taskDTO, Object data) {
        HtScanningMetadataDTO metadataDTO = (HtScanningMetadataDTO)data;
        List<HtScanningMetadataDTO> dataList = new ArrayList<>();
        dataList.add(metadataDTO);
        AlgorithmTaskRuleParam taskRuleParam = JSON.parseObject(JSON.toJSONString(ruleParams), AlgorithmTaskRuleParam.class);
        AlgorithmTaskPathParam taskPathParam = JSON.parseObject(JSON.toJSONString(inputParams), AlgorithmTaskPathParam.class);
        assemble(taskRuleParam,taskPathParam,dataList,taskDTO);
    }
    /**
     * 校验输入规则的文件是否写入路径
     * @param ruleParam
     * @param pathParam
     * @return
     */
    private boolean rulePathExist(AlgorithmTaskRuleParam ruleParam, AlgorithmTaskPathParam pathParam) {
        if (StringUtils.isBlank(ruleParam.getPrimaryFile())) {
            logger.info("主文件规则为空");
            return false;
        }
        if (StringUtils.isNotBlank(ruleParam.getGeoFile())) {
            RuleVO geoRuleVO = JSON.parseObject(ruleParam.getGeoFile(), RuleVO.class);
            if (!NULL.equals(geoRuleVO.getMatchRule()) && StringUtils.isBlank(pathParam.getGeoFile())) {
                logger.info("geo文件输入路径为空");
                return false;
            }
        }
        if (StringUtils.isNotBlank(ruleParam.getL2File())) {
            RuleVO l2RuleVO = JSON.parseObject(ruleParam.getL2File(), RuleVO.class);
            if (!NULL.equals(l2RuleVO.getMatchRule()) && StringUtils.isBlank(pathParam.getL2File())) {
                logger.info("L2文件输入路径为空");
                return false;
            }
        }
        if (StringUtils.isNotBlank(ruleParam.getNwpFile())) {
            RuleVO nwpRuleVO = JSON.parseObject(ruleParam.getNwpFile(), RuleVO.class);
            if (!NULL.equals(nwpRuleVO.getMatchRule()) && StringUtils.isBlank(pathParam.getNwpFile())) {
                logger.info("辅助（nwp）文件输入路径为空");
                return false;
            }
        }
        return true;
    }


    protected List<HtScanningMetadataDTO> selectExecMainFileList(RuleVO primaryRuleVO, boolean preprocess,
                                                        HtScheduleJobDTO taskDTO) {
        //获取主波段或传感器参数（）
        if (StringUtils.isBlank(primaryRuleVO.getMatchRule()) || NULL.equals(primaryRuleVO.getMatchRule())) {
            new ArrayList<>();
        }
        String mainMathexp = primaryRuleVO.getMatchRule().split(",|\\|")[0];
        String productIdentify = primaryRuleVO.getProductIdentify();
        String taskType = taskDTO.getTaskType();
        LocalDateTime now = LocalDateTime.now();
        String startTime = TIME_FORMAT.format(now.plusDays(-1L * nearDay));
        String endTime = TIME_FORMAT.format(now);
        if ("redo_singleTask".equals(taskType)) {
            Date redoStartTime = taskDTO.getRedoStartTime();
            Date redoEndTime = taskDTO.getRedoEndTime();
            startTime = TIME_FORMAT.format(redoStartTime.toInstant().atZone(ZoneId.systemDefault()));
            endTime = TIME_FORMAT.format(redoEndTime.toInstant().atZone(ZoneId.systemDefault()));
        }
        List<HtScanningMetadataDTO> allMetaDataList = null;
        /*
            预处理：根据传入的传感器，时间范围，数据等级查询。
            专业处理：根据传入的通道，时间范围，文件类型（L2X、L2Y的文件类型为TIFF），数据等级查询。
         */
        if (preprocess) {
            allMetaDataList = htScanningMetadataService.listBySensorAndRangeTimeAndLevel(mainMathexp,startTime,endTime,primaryRuleVO.getLevel());
        } else {
            allMetaDataList = htScanningMetadataService.listByBandsAndDataLevelAndProductIidAndRangeTime(mainMathexp,primaryRuleVO.getLevel(),productIdentify,startTime,endTime);
        }
        List<String> allNameList = allMetaDataList.stream().map(p->p.getFileName()).collect(Collectors.toList());
//        List<DbDmsSchedulerTaskFileRecordDTO> existRecordDataList = dbDmsSchedulerTaskFileRecordService.findByTaskId(taskDTO.getId());
        List<DbDmsSchedulerTaskFileRecordDTO> existRecordDataList = dbDmsSchedulerTaskFileRecordService.findByTaskIdAndFileNameIn(taskDTO.getId(),allNameList);
        List<HtScanningMetadataDTO> execMetadataList = null;
        //过滤需要执行的主文件，如果记录表中为空则查询出的时间范围内所有文件都执行，如果不为空，则筛选掉已经执行过的。
        if (existRecordDataList.isEmpty()) {
            execMetadataList = allMetaDataList;
        } else {
            Map<String, DbDmsSchedulerTaskFileRecordDTO> hash = existRecordDataList.stream().
                    collect(Collectors.toMap(DbDmsSchedulerTaskFileRecordDTO::getFileName, dto -> dto, (v1, v2) -> v2));
            execMetadataList = allMetaDataList.stream().filter(dto -> !hash.containsKey(dto.getFileName())).collect(Collectors.toList());
        }
        return execMetadataList;
    }

    protected List<HtScanningMetadataDTO> findRedoMetadataList(String taskId) {
        List<DbDmsSchedulerTaskFileRecordDTO> redoTaskRecordList = dbDmsSchedulerTaskFileRecordService.findRedoTasks(taskId);
        List<String> fileNameList = redoTaskRecordList.stream()
                .map(DbDmsSchedulerTaskFileRecordDTO::getFileName).collect(Collectors.toList());
        return htScanningMetadataService.findByFileNameList(fileNameList);
    }

    private String primaryRule(RuleVO primaryFileRule, boolean preprocess, HtScanningMetadataDTO mainFileDTO,
                               HtPathConfigurationDTO pathDTO, String matchRule) {
        //主文件规则时间范围
        String mainFileDate = mainFileDTO.getFileDate();
        Map<String,String> timeRangeParam = parseTimeRange(mainFileDate, primaryFileRule.getTimeRange());
        //查找满足要求的所有文件列表
        String[] mathexpArr = matchRule.split("\\|");
        String leftTime = timeRangeParam.get("startTime");
        String rightTime = timeRangeParam.get("endTime");
        //如果没有输入时间参数，则默认考虑正负3分钟查找文件，经纬度满足正负0.5°
        if (leftTime.equals(rightTime)) {
            return findPrimaryFileList(mainFileDTO, primaryFileRule.getLevel(), pathDTO, Arrays.asList(mathexpArr), preprocess);
        } else {
            //1、根据主通道找到输入的时间范围内的所有主通道文件。2、根据不同时次的主通道正负3分钟找到同一谱段的其他通道文件，拼接
            List<HtScanningMetadataDTO> mainBandsFileList = htScanningMetadataService
                    .listByBandsAndDataLevelAndRangeTime(mainFileDTO.getBands(),primaryFileRule.getLevel(),leftTime,rightTime);
            if (mainBandsFileList.isEmpty()) {
                logger.info("未查询到多个谱段的主通道文件时");
                return null;
            }
            StringBuilder primaryFilePath = new StringBuilder();
            for (HtScanningMetadataDTO dto : mainBandsFileList) {
                String path = findPrimaryFileList(dto, primaryFileRule.getLevel(), pathDTO, Arrays.asList(mathexpArr), preprocess);
                if (path == null) {
                    return null;
                }
                primaryFilePath.append(";").append(path);
            }
            return primaryFilePath.substring(1);
        }
    }

    /**
     * 获取primaryFile文件列表信息
     * @param mainFileDTO
     * @param level
     * @param bandsList
     * @return
     */
    private String findPrimaryFileList(HtScanningMetadataDTO mainFileDTO, String level, HtPathConfigurationDTO pathDTO,
                                        List<String> bandsList,boolean preProcess) {
        List<HtScanningMetadataDTO> resultList = null;
        //标准化处理
        if (preProcess) {
            resultList = new ArrayList<>();
            resultList.add(mainFileDTO);
        } else {
            String mainFileDate = mainFileDTO.getFileDate();
            String leftTime = TIME_FORMAT.format(LocalDateTime.parse(mainFileDate,TIME_FORMAT).plusMinutes(-1 * DIFF_TIME_MINUTE));
            String rightTime = TIME_FORMAT.format(LocalDateTime.parse(mainFileDate,TIME_FORMAT).plusMinutes(DIFF_TIME_MINUTE));
            List<HtScanningMetadataDTO> dataList = htScanningMetadataService
                    .listByBandsAndDataLevelRangeTime(bandsList,level,leftTime,rightTime);
            if (dataList.isEmpty()) {
                return null;
            }
            double lon = Double.parseDouble(mainFileDTO.getLon().substring(1));
            double lat = Double.parseDouble(mainFileDTO.getLat().substring(1));
            resultList = dataList.stream().filter(dto -> {
                double dLon = Double.parseDouble(dto.getLon().substring(1));
                double dLat = Double.parseDouble(dto.getLat().substring(1));
                return dLon >= lon - 0.5 && dLon <= lon + 0.5 && dLat >= lat - 0.5 && dLat <= lat + 0.5;
            }).collect(Collectors.toList());
            //校验查询出的数据是否和输入通道数量匹配
            Map<String, List<HtScanningMetadataDTO>> hash = resultList.stream().collect(Collectors.groupingBy(HtScanningMetadataDTO::getBands));
            if (hash.size() != bandsList.size()) {
                logger.info("查询出的通道数量与输入数量不匹配");
                return null;
            }
            List<HtScanningMetadataDTO> tempList = new ArrayList<>();
            long mainFileMilli = LocalDateTime.parse(mainFileDate, TIME_FORMAT).atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();
            hash.values().forEach(list -> {
                HtScanningMetadataDTO resultDTO = list.stream().min((a, b) -> {
                    long aMilli = LocalDateTime.parse(a.getFileDate(), TIME_FORMAT).atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();
                    long bMilli = LocalDateTime.parse(b.getFileDate(), TIME_FORMAT).atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();
                    return Math.abs(aMilli - mainFileMilli) > Math.abs(bMilli - mainFileMilli) ? 1 : -1;
                }).get();
                tempList.add(resultDTO);
            });
            resultList = tempList;
        }
        //判断磁盘上是否存在文件
        StringBuilder sb = new StringBuilder();
        for (HtScanningMetadataDTO dto : resultList) {
            String currentFilePath = pathDTO.getClientPath() + File.separator + dto.getFilePath();
            File file = new File(currentFilePath);
            if (!file.exists()) {
                return null;
            }
            sb.append(",").append(currentFilePath);
        }
        return sb.substring(1);
    }

    private Map<String,String> geoRule(String geoRule, boolean preProcess, HtScanningMetadataDTO mainFileDTO, String geoPathPrefix) {
        StringBuilder geoPathSb = new StringBuilder();
        StringBuilder rpbPathSb = new StringBuilder();
        Map<String,String> resultMap = new HashMap<>();
        resultMap.put("geo","");
        resultMap.put("rpb","");
        if (StringUtils.isBlank(geoRule)) {
            return resultMap;
        }
        RuleVO geoRuleVO = JSON.parseObject(geoRule, RuleVO.class);
        if (NULL.equals(geoRuleVO.getMatchRule()) || NULL.equals(geoRuleVO.getLevel())) {
            return resultMap;
        }
        Map<String,String> timeRangeParam = parseTimeRange(mainFileDTO.getFileDate(), geoRuleVO.getTimeRange());
        String[] mathexpArr = geoRuleVO.getMatchRule().split("\\|");
        List<HtScanningMetadataDTO> ruleFileList = null;
        //标准化处理根据传感器，时间范围，数据等级，查询
        String startTime = timeRangeParam.get("startTime");
        String endTime = timeRangeParam.get("endTime");
        if (preProcess) {
            ruleFileList = htScanningMetadataService.listBySensorAndRangeTimeAndLevelAndBand(mainFileDTO.getSensor(),startTime,
                    endTime,geoRuleVO.getLevel(), mainFileDTO.getBands());
            Set<String> mathexpHashSet = new HashSet<>(Arrays.asList(mathexpArr));
            ruleFileList = ruleFileList.stream().filter(item -> {
                String name = item.getFileName();
                String[] splitName = name.split("_");
                return mathexpHashSet.contains(splitName[splitName.length - 2]);
            }).collect(Collectors.toList());
            if (ruleFileList.size() != mathexpHashSet.size()) {
                logger.info("主文件{}扫描过程中，查询geo文件和rpb文件时输入参数和查询出的数量不一致",mainFileDTO.getFileName());
                return null;
            }
        } else {
            //TODO 专业处理时HTHTGEO扫描规则
//            ruleFileList = htScanningMetadataService.listByProductIidAndBandsAndDataLevelAndLonAndLatAndTimeRange(mathexpArr[0],mainFileDTO.getBands(),
//                    geoRuleVO.getLevel(),mainFileDTO.getLon(),mainFileDTO.getLat(),startTime,endTime);
        }
        if (ruleFileList.isEmpty()) {
            logger.info("未查询到任何geo或rgb文件");
            return null;
        }
        for (HtScanningMetadataDTO fileDTO : ruleFileList) {
            String currentFilePath = geoPathPrefix + File.separator + fileDTO.getFilePath();
            File file = new File(currentFilePath);
            if (!file.exists()) {
                logger.info( "{}在磁盘上不存在",fileDTO.getFileName());
                return null;
            }
            if ("HDF".equals(fileDTO.getFileType())) {
                geoPathSb.append(",").append(currentFilePath);
            } else if ("RPB".equals(fileDTO.getFileType())) {
                rpbPathSb.append(",").append(currentFilePath);
            }
        }
        String geoPath = geoPathSb.toString();
        String rpbPath = rpbPathSb.toString();
        if (StringUtils.isNotBlank(geoPath)) {
            resultMap.put("geo",geoPath.substring(1));
        }
        if (StringUtils.isNotBlank(rpbPath)) {
            resultMap.put("rpb",rpbPath.substring(1));
        }
        return resultMap;
    }

    private String l2Rule(String rule, HtScanningMetadataDTO mainFileDTO,String l2PathPrefix) {
        StringBuilder filePathSb = new StringBuilder();
        if (StringUtils.isBlank(rule)) {
            return filePathSb.toString();
        }
        RuleVO l2RuleVO = JSON.parseObject(rule, RuleVO.class);
        if (NULL.equals(l2RuleVO.getMatchRule()) || NULL.equals(l2RuleVO.getLevel())) {
            return filePathSb.toString();
        }
        Map<String,String> timeRangeParam = parseTimeRange(mainFileDTO.getFileDate(), l2RuleVO.getTimeRange());
        String[] mathexpArr = l2RuleVO.getMatchRule().split("\\|");
        List<HtScanningMetadataDTO> ruleFileList;
        String startTime = timeRangeParam.get("startTime");
        String endTime = timeRangeParam.get("endTime");
        boolean isSingleTimes = false;
        if (startTime.equals(endTime)) {
            isSingleTimes = true;
            startTime = TIME_FORMAT.format(LocalDateTime.parse(mainFileDTO.getFileDate(),TIME_FORMAT).plusMinutes(-1 * DIFF_TIME_MINUTE));
            endTime = TIME_FORMAT.format(LocalDateTime.parse(mainFileDTO.getFileDate(),TIME_FORMAT).plusMinutes(DIFF_TIME_MINUTE));
            List<HtScanningMetadataDTO> dataList = htScanningMetadataService
                    .listByProductIidListAndTimeRangeAndLevel(Arrays.asList(mathexpArr), startTime,endTime,l2RuleVO.getLevel());
            double lon = Double.parseDouble(mainFileDTO.getLon().substring(1));
            double lat = Double.parseDouble(mainFileDTO.getLat().substring(1));
            ruleFileList = dataList.stream().filter(dto -> {
                double dLon = Double.parseDouble(dto.getLon().substring(1));
                double dLat = Double.parseDouble(dto.getLat().substring(1));
                return dLon >= lon - 0.5 && dLon <= lon + 0.5 && dLat >= lat - 0.5 && dLat <= lat + 0.5;
            }).collect(Collectors.toList());
        } else {
            //TODO L2多时次文件查询规则，暂时用不到
            logger.info("多时次查询");
            ruleFileList = new ArrayList<>();
        }
        Map<String, List<HtScanningMetadataDTO>> filterMap = ruleFileList.stream()
                .collect(Collectors.groupingBy(HtScanningMetadataDTO::getProductIid));
        if (filterMap.size() != mathexpArr.length) {
            logger.info("与主文件{}要求的L2级文件数量与L2级规则产品标识数量不一致",mainFileDTO.getFileName());
            return null;
        }
        if (isSingleTimes) {
            List<HtScanningMetadataDTO> tempList = new ArrayList<>();
            long mainFileMilli = LocalDateTime.parse(mainFileDTO.getFileDate(), TIME_FORMAT).atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();
            filterMap.values().forEach(list -> {
                HtScanningMetadataDTO resultDTO = list.stream().min((a, b) -> {
                    long aMilli = LocalDateTime.parse(a.getFileDate(), TIME_FORMAT).atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();
                    long bMilli = LocalDateTime.parse(b.getFileDate(), TIME_FORMAT).atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();
                    return Math.abs(aMilli - mainFileMilli) > Math.abs(bMilli - mainFileMilli) ? 1 : -1;
                }).get();
                tempList.add(resultDTO);
            });
            ruleFileList = tempList;
        }
        for (HtScanningMetadataDTO fileDTO : ruleFileList) {
            String currentPath = l2PathPrefix + File.separator + fileDTO.getFilePath();
            File file = new File(currentPath);
            if (!file.exists()) {
                return null;
            }
            filePathSb.append(",").append(currentPath);
        }
        return filePathSb.substring(1);
    }

    private String nwpRule(String nwpFile, String mainFileDate, String npwPathPrefix) {
        StringBuilder filePathSb = new StringBuilder();
        if (StringUtils.isBlank(nwpFile)) {
            return filePathSb.toString();
        }
        RuleVO nwpRuleVO = JSON.parseObject(nwpFile, RuleVO.class);
        if (NULL.equals(nwpRuleVO.getMatchRule()) || NULL.equals(nwpRuleVO.getLevel())) {
            return filePathSb.toString();
        }
        //将主文件时间转为世界时来匹配数值预报数据
        mainFileDate = TIME_FORMAT.format(LocalDateTime.parse(mainFileDate, TIME_FORMAT).plusHours(-8L));
        //查询48小时内在数据库种条件要素的文件是否存在，并且数量至少为3个。
        Long earlyHour = (-1) * 48L;
        String earlyTime = TIME_FORMAT.format(LocalDateTime.parse(mainFileDate, TIME_FORMAT).plusHours(earlyHour));
        String[] mathexpArr = nwpRuleVO.getMatchRule().split("\\|");
        List<DbDmsScanningMetadataAuxInfoDTO> earlyAllData = dbDmsScanningMetadataAuxInfoService.
                listByFileWantListAndRangeTimeOrder(Arrays.asList(mathexpArr), earlyTime, mainFileDate);
        Map<String, Long> countAllNwpMap = earlyAllData.stream()
                .collect(Collectors.groupingBy(DbDmsScanningMetadataAuxInfoDTO::getFileWant, Collectors.counting()));
        if (mathexpArr.length != countAllNwpMap.size()) {
            logger.info("48小时内满足的要素种类数量不正确");
            return null;
        }
        for (Map.Entry<String, Long> entry : countAllNwpMap.entrySet()) {
            if (entry.getValue() < 3) {
                logger.info("{}--->要素的数量在48小时的范围内小于3", entry.getKey());
            }
        }
        //过滤筛筛选出符合规则要求的时次的要素文件  每种要素3个数据信息
        Map<String,String> timeRangeParam = parseTimeRange(mainFileDate, nwpRuleVO.getTimeRange());
        LocalDateTime nwpStartTime = LocalDateTime.parse(timeRangeParam.get("startTime"), TIME_FORMAT);
        LocalDateTime nwpEndTime = LocalDateTime.parse(timeRangeParam.get("endTime"), TIME_FORMAT);
        //key:要素 value:[nwp数据（时间+时次会重复）]
        Map<String, List<DbDmsScanningMetadataAuxInfoDTO>> filterMap = earlyAllData.stream().filter(dto -> {
            LocalDateTime dtoLocalDateTime = LocalDateTime.parse(dto.getFileDate(), TIME_FORMAT).plusHours(dto.getFileTimeSecond());
            return !nwpStartTime.isAfter(dtoLocalDateTime) && !nwpEndTime.isBefore(dtoLocalDateTime);
        }).collect(Collectors.groupingBy(DbDmsScanningMetadataAuxInfoDTO::getFileWant));
        for (Map.Entry<String,List<DbDmsScanningMetadataAuxInfoDTO>> entry : filterMap.entrySet()) {
            List<DbDmsScanningMetadataAuxInfoDTO> wantList = entry.getValue();
            List<DbDmsScanningMetadataAuxInfoDTO> nwpInfoList = new ArrayList<>();
            Map<String,DbDmsScanningMetadataAuxInfoDTO> hash = new HashMap<>();
            for (DbDmsScanningMetadataAuxInfoDTO nwpDTO : wantList) {
                String formatDate = TIME_FORMAT.format(LocalDateTime.parse(nwpDTO.getFileDate(),TIME_FORMAT).plusHours(nwpDTO.getFileTimeSecond()));
                if (!hash.containsKey(formatDate)) {
                    hash.put(formatDate,nwpDTO);
                    nwpInfoList.add(nwpDTO);
                }
            }
            if (nwpInfoList.size() < 3) {
                logger.info("{}要素在规则的时间范围内个数小于3",entry.getKey());
                return null;
            }
            //查找最接近的三个数量文件
            List<Long> nearPrimaryFileTime = findNearDateNwpFile(nwpInfoList,mainFileDate);
            //判断磁盘上是否存在
            StringBuilder sb = new StringBuilder();
            for (Long keyMilli : nearPrimaryFileTime) {
                String keyTime = TIME_FORMAT.format(LocalDateTime.ofInstant(Instant.ofEpochMilli(keyMilli),ZoneId.systemDefault()));
                DbDmsScanningMetadataAuxInfoDTO currentFileDTO = hash.get(keyTime);
                String currentPath = npwPathPrefix + File.separator + currentFileDTO.getFilePath();
                File file = new File(currentPath);
                if (!file.exists()) {
                    logger.info("{}--->nwp文件在磁盘上不存在",file.getName());
                    return null;
                }
                sb.append(",").append(currentPath);
            }
            filePathSb.append(";").append(sb.substring(1));
        }
        return filePathSb.substring(1);
    }

    private List<Long> findNearDateNwpFile(List<DbDmsScanningMetadataAuxInfoDTO> nwpInfoList, String mainFileDate) {
        long primaryFileMilli = LocalDateTime.parse(mainFileDate,TIME_FORMAT).atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();
        List<Long> nearPrimaryFileTime = new ArrayList<>();
        for (int j = 0; j < nwpInfoList.size(); j++) {
            DbDmsScanningMetadataAuxInfoDTO metadataAuxInfoDTO = nwpInfoList.get(j);
            long currentFileMilli = LocalDateTime.parse(metadataAuxInfoDTO.getFileDate(),TIME_FORMAT)
                    .plusHours(metadataAuxInfoDTO.getFileTimeSecond())
                    .atZone(ZoneId.systemDefault())
                    .toInstant()
                    .toEpochMilli();
            if (j < 3) {
                nearPrimaryFileTime.add(currentFileMilli);
            } else {
                //主文件时次右边是从大到小遍历，所以直接替换，保持最右是最小最近的文件时间，最左边是最大的。
                if (currentFileMilli - primaryFileMilli >= 0) {
                    nearPrimaryFileTime.remove(0);
                    nearPrimaryFileTime.add(currentFileMilli);
                } else {
                    //左边进行比较
                    long dif = primaryFileMilli - currentFileMilli;
                    long max = nearPrimaryFileTime.stream().max((a,b) -> Math.abs(a - primaryFileMilli) > Math.abs(b - primaryFileMilli) ? 1 : -1).get();
                    if (dif < Math.abs(max - primaryFileMilli)) {
                        nearPrimaryFileTime.remove(max);
                        nearPrimaryFileTime.add(currentFileMilli);
                    } else {
                        break;
                    }
                }
            }
        }
        return nearPrimaryFileTime;
    }


    private Map<String,String> parseTimeRange(String mainFileDate,String timeRange) {
        Map<String,String> resultMap = new HashMap<>();
        resultMap.put("startTime",mainFileDate);
        resultMap.put("endTime",mainFileDate);
        if (timeRange.equals("NULL")) {
            return resultMap;
        }
        String[] timeSplit = timeRange.split(",");
        long startPlusMinute = -1 * Math.abs(Long.parseLong(timeSplit[0]));
        long endPlusMinute = Math.abs(Long.parseLong(timeSplit[1]));
        LocalDateTime mainFileLocalDateTime = LocalDateTime.parse(mainFileDate, TIME_FORMAT);
        String startTime = TIME_FORMAT.format(mainFileLocalDateTime.plusMinutes(startPlusMinute));
        String endTime = TIME_FORMAT.format(mainFileLocalDateTime.plusMinutes(endPlusMinute));
        resultMap.put("startTime",startTime);
        resultMap.put("endTime",endTime);
        return resultMap;
    }

    /**
     * 修改任务记录信息并发送到指定TOPIC消息，需要在每次扫描并组装好数据后调用
     * @param execParams 所有额外参数
     * @param taskDTO 任务信息对象
     * @param fileName 主文件
     * @param algorithmExecParams 算法执行参数
     * @return
     */
    @Transactional(rollbackFor = Exception.class)
    public void updateRecordAndSendMessageInsertFlow(Map<String,Object> execParams, HtScheduleJobDTO taskDTO, String fileName,
                                           Map<String,Object> algorithmExecParams) {
        String mainFileName = fileName;
        String taskPlanId = (String) execParams.get("taskPlanId");
        //添加记录信息
        DbDmsSchedulerTaskFileRecordDTO record = dbDmsSchedulerTaskFileRecordService.findByTaskIdAndFileName(taskDTO.getId(), mainFileName);
        if (record == null) {
            record = dbDmsSchedulerTaskFileRecordService.prepareDataByFileName(taskDTO.getId(),mainFileName,taskPlanId);
        } else {
            dbDmsSchedulerTaskFileRecordService.dataRedo(record);
            dbDmsSchedulerTaskRecordLogService.deleteByRid(record.getId());
            dbDmsSchedulerTaskFileRecordFlowService.deleteByRid(record.getId());
        }
        //查询表中必备字段
        HtAlgorithmConfigurationDTO algorithmDTO = htAlgorithmConfigurationService.findById(taskDTO.getAlgoId());
        Map<String,Object> resultMap = new HashMap<>();
        resultMap.put("algorithmParameter",algorithmExecParams);
        Map<String,Object> basic = new HashMap<>();
        basic.put("algorithmIdentify",algorithmDTO.getAlgorithmIdentify());
        basic.put("hasFlow",false);
        basic.put("taskId",taskDTO.getId());
        basic.put("fileTaskID",record.getId());
        basic.put("mainFileName",fileName);
        resultMap.put("basic",basic);
        resultMap.put("other",execParams);
        //向景号记录表里增加记录
        insertIntoViewNumberRecord(resultMap,record);

        kafkaTemplate.send(execTopic,JSON.toJSONString(resultMap));


    }
    /**
     * @Description:向景号记录表里增加数据
     * @Author: zyp
     * @Date: 2021/5/28
     * @param: []
     * @return: void
     **/
    public void insertIntoViewNumberRecord(Map<String,Object> resultMap,DbDmsSchedulerTaskFileRecordDTO record){
        Map<String,Object> basicMap = (Map<String, Object>) resultMap.get("basic");
        Map<String,Object> otherMap = (Map<String, Object>) resultMap.get("other");
        String viewNumber = (String) otherMap.get("viewNumber");
        String taskId = (String) basicMap.get("taskId");
        String taskPlanId = (String) otherMap.get("taskPlanId");
        String recordId = record.getId();
        String fileName = (String) basicMap.get("mainFileName");
        String algorithmid = (String) otherMap.get("algorithmId");
        HtScanningMetadataDTO htScanningMetadataDTO =  htScanningMetadataService.findByFileName(fileName);
        HtScheduleJobDTO htScheduleJobDTO = htScheduleJobService.findById(taskId);
        String satellite = htScanningMetadataDTO.getSatellite();
        String flowId = htScheduleJobDTO.getFlowId();
        if(StringUtils.isNotBlank(flowId)){
            //根据景号查询任务执行流程信息表是否有数据，有，向子表插入主文件信息；没有，增加一条景号记录，再增加一条子表数据
            DbDmsFlowInfoExecDTO execDTO = dbDmsFlowInfoExecService.findByViewNumber(viewNumber);
            if(execDTO == null){
                execDTO = new DbDmsFlowInfoExecDTO();
                execDTO.setSatellite(satellite);
                execDTO.setTaskPlanId(taskPlanId);
                execDTO.setViewNumber(viewNumber);
                execDTO.setCreatetime(new java.util.Date());
                execDTO = dbDmsFlowInfoExecService.save(execDTO);
            }
            DbDmsFlowInfoExecStatusDTO dbDmsFlowInfoExecStatusDTO = new DbDmsFlowInfoExecStatusDTO();
            dbDmsFlowInfoExecStatusDTO.setAlgoId(algorithmid);
            dbDmsFlowInfoExecStatusDTO.setExecId(execDTO.getId());
            dbDmsFlowInfoExecStatusDTO.setRecordId(recordId);
            dbDmsFlowInfoExecStatusService.save(dbDmsFlowInfoExecStatusDTO);
            //如果有景号记录，根据taskId查询流程id集合，根据流程id和景号记录表主键 查询(流程信息执行关系表) 是否有数据，有不增加，没有增加
            String[] flowIds = flowId.split(",");
            if(execDTO != null){
                for(String newFlowId:flowIds){
                    List<DbDmsFlowInfoExecIdFlowIdDTO> defd  = dbDmsFlowInfoExecIdFlowIdService. findByFlowIdAndExecId(newFlowId,execDTO.getId());
                    if(defd.size()<1 && defd.isEmpty()){
                        DbDmsFlowInfoExecIdFlowIdDTO def = new DbDmsFlowInfoExecIdFlowIdDTO();
                        java.util.Date date = new java.util.Date();
                        def.setExecId(execDTO.getId());
                        def.setFlowId(newFlowId);
                        def.setState(0);
                        dbDmsFlowInfoExecIdFlowIdService.saveAndFlush(def);
                        ViewRecordVo vo = new ViewRecordVo();
                        vo.setExecId(execDTO.getId());
                        vo.setFlowId(newFlowId);
                        vo.setSatellite(satellite);
                        vo.setState(0);
                        vo.setTaskPlanId(taskPlanId);
                        vo.setViewNumber(viewNumber);
                        vo.setCreatetime(date);
                        SimpleDateFormat sdf = new SimpleDateFormat("HHmmss");
                        String dat = sdf.format(date);
                        double score = Double.parseDouble(dat);
                        //往redis中存储数据,半小时后删除
                        Jedis jedis = RedisUtil.getJedis();
                        try {
                            if (jedis != null) {
                                String recordKey = String.format("viewNumber:%s:%s",newFlowId,viewNumber);
                                jedis.set(recordKey,JSON.toJSONString(vo));
                                jedis.expire(recordKey,3600*24);
                                //value为recordKey的索引，sorce为时分秒转为的double
                                redisTemplate.opsForZSet().add("viewNumberRecord:"+newFlowId,recordKey,score);
                                redisTemplate.expire("viewNumberRecord:"+newFlowId,3600*24, TimeUnit.SECONDS);
                            }
                        } finally {
                            RedisUtil.close(jedis);
                        }
                    }
                }

            }
        }
    }
}
